package com.burges.net.tableAPIAndSQL.sql

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{Table, TableEnvironment, Types}
import org.apache.flink.table.sinks.CsvTableSink

/**
  * 创建人    BurgessLee
  * 创建时间   2020/2/25
  * 描述     FLink SQL应用实例
  */
object SqlDEmo {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		// 获取TableEnvironment对象
		val tableEnvironment = TableEnvironment.getTableEnvironment(environment)
		// 构建数据集
		val dataStream = environment.fromElements(("1","flink"))
		val sensors_table: Table = tableEnvironment.fromDataStream(dataStream)
		//表结构 id, type, timestamp, var1, var2
		tableEnvironment.registerTable("sensors", sensors_table)
		val csvTableSink = new CsvTableSink("/path/file/csvfile")
		//定义字段名称
		val fieldName: Array[String] = Array("id", "type")
		//定义字段类型
		val fieldTypes: Array[TypeInformation[_]] = Array(Types.LONG, Types.STRING)
		//通过registerTableSink将csvTableSink注册成Table
		tableEnvironment.registerTableSink("csv_output_file", fieldName, fieldTypes, csvTableSink)
		//计算与传感器类型为A的每个传感器id对应的var1指标的和
		val rs: Table = tableEnvironment.sqlQuery("select id, sum(var1) as sumvar1 from sensors_table where type='speed' group by sensor_id")
		//通过SqlUpdate方法将类型为温度的数据筛选出来并输出到外部表中
		tableEnvironment.sqlUpdate("insert into csv_output_file select product, amount from Sensors where type='temperature'")
	}

}
